package org.systemsbiology.rface.hadoop.data;

import org.apache.commons.math.util.FastMath;
import org.apache.hadoop.io.Writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.PriorityQueue;

/**
 * Created by IntelliJ IDEA.
 * User: anorberg
 * Date: 9/15/11
 * Time: 3:27 PM
 */
public class ResultGroup implements Writable {

    private double[] values;
    private double[] targets;


    private static final int MAGIC_SIGNATURE_OPEN = 0xF00DBEEF;
    private static final int MAGIC_SIGNATURE_CLOSE = 0xFADEBABE;

    /**
     * Creates an empty ResultGroup. Useless except for readFields.
     */
    public ResultGroup(){
        values = targets = null;
    }

    /**
     * Creates a ResultGroup representing a single result. Merge ResultGroups to get larger groups.
     * @param value The strength of their correlation.
     * @param isTarget Whether or not this particular value is in a file we're just adding and are
     *        thus analyzing as a potential outlier.
     */
    public ResultGroup(double value, boolean isTarget){
    	value = unlog(value);
        if(isTarget){
            targets = new double[1];
            targets[0] = value;
            values = new double[0];
        } else {
            targets = new double[0];
            values = new double[1];
            values[0] = value;
        }
    }
    
    public ResultGroup(ResultGroup dupeMe){
    	targets = Arrays.copyOf(dupeMe.targets, dupeMe.targets.length);
    	values = Arrays.copyOf(dupeMe.values, dupeMe.values.length);
    }
    
    /**
     * Helper function. Some data sets have had a LOG10 transform on the data column, some haven't.
     * The negative numbers are a dead giveaway.
     * @param Value in the data column.
     * @return The real data value behind it- either that value, or 10^that value.
     */
    private static double unlog(double x){
        if(x >= 0) return x;
        return FastMath.pow(10, x);
    }

    public double[] targetsArray() {
        return targets;
    }

    public double[] valuesArray(){
        return values;
    }

    private static class DoubleMerge implements Comparable<DoubleMerge>{
        private double[] values;
        private int index;

        public DoubleMerge(double[] data){
            index = 0;
            values = data;
        }

        public double take(){
            return values[index++];
        }

        public boolean remain(){
            return index < values.length;
        }

        public int compareTo(DoubleMerge d){
        	return Double.compare(values[index], d.values[d.index]);
        }
        
        public boolean equals(Object o){
        	if(o == null) return false;
        	if(!o.getClass().equals(getClass())) return false;
        	DoubleMerge dm = (DoubleMerge) o;
        	if(index != dm.index) return false;
        	return Arrays.equals(values, dm.values);
        }
        
        public int hashCode(){
        	int x = Arrays.hashCode(values);
        	return x << (index % 32) + x >>> (index % 32);
        }
    }

    public static ResultGroup merge(Iterable<ResultGroup> groups){
        PriorityQueue<DoubleMerge> pq = new PriorityQueue<DoubleMerge>();

        ArrayList<ResultGroup> groupList= new ArrayList<ResultGroup>();
        int valueTotal = 0;
        for(ResultGroup g : groups){
        	g = new ResultGroup(g);
            groupList.add(g);
            if(g.values.length > 0){
                pq.add(new DoubleMerge(g.values));
                valueTotal += g.values.length;
            }
        }

        if(groupList.size() == 0){
            return null;
        }

        double[] values = new double[valueTotal];
        int k = 0;
        while(!pq.isEmpty()){
            DoubleMerge rec = pq.poll();
            values[k++] = rec.take();
            if(rec.remain()){
                pq.add(rec);
            }
        }

        if(k != valueTotal){
            throw new AssertionError("Must have dropped a result on the floor somewhere- didn't fill values");
        }

        //at this point, pq is empty so we can just recycle it
        int targetTotal = 0;
        for(ResultGroup g: groupList){
            if(g.targets.length > 0){
                pq.add(new DoubleMerge(g.targets));
                targetTotal += g.targets.length;
            }
        }

        double[] targets = new double[targetTotal];
        k = 0;
        while(!pq.isEmpty()){
            DoubleMerge rec = pq.poll();
            targets[k++] = rec.take();
            if(rec.remain()){
                pq.add(rec);
            }
        }

        if(k != targetTotal){
            throw new AssertionError("Must have dropped a target on the floor somewhere- didn't fill targets");
        }

        ResultGroup ret = new ResultGroup();
        ret.values= values;
        ret.targets = targets;
        return ret;
    }

    public int valuesCount(){
        return values.length;
    }

    public double getValue(int dex){
        return values[dex];
    }

    public int targetsCount(){
        return targets.length;
    }

    public double getTarget(int dex){
        return targets[dex];
    }

    //TODO: stddev
    //TODO: mean
    //TODO: logitnormal sigma
    //TODO: peakiness
    //TODO: prob dens fxn plotter

    // ============= DATA FORMAT: ===================
    // the int 0xF00DBEEF
    // int - number of following double records
    // sequence of doubles- a values[], of length (previous int)
    // int - number of following double records
    // sequence of doubles - a targets[], of length (previous int)
    // the int 0xFADEBABE

    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeInt(MAGIC_SIGNATURE_OPEN);

        dataOutput.writeInt(values.length);
        for(double d:values){
            dataOutput.writeDouble(d);
        }

        dataOutput.writeInt(targets.length);
        for(double d:targets){
            dataOutput.writeDouble(d);
        }

        dataOutput.writeInt(MAGIC_SIGNATURE_CLOSE);
    }

    public void readFields(DataInput dataInput) throws IOException {
        if(dataInput.readInt() != MAGIC_SIGNATURE_OPEN){
            throw new IOException("Opening signature failed to match! Not at a ResultGroup record!");
        }

        int targetLen = dataInput.readInt();
        if(values == null || values.length != targetLen){
            values = new double[targetLen];
        }
        for(int k = 0; k < values.length; ++k){
            values[k] = dataInput.readDouble();
        }

        targetLen = dataInput.readInt();
        if(targets == null || targets.length != targetLen){
            targets = new double[targetLen];
        }
        for(int k = 0; k < targets.length; ++k){
            targets[k] = dataInput.readDouble();
        }

        if(dataInput.readInt() != MAGIC_SIGNATURE_CLOSE){
            throw new IOException("Closing signature failed to match! Invalid record length parameters?");
        }

        try{
            verify();
        } catch (IllegalStateException wrongo){
            throw new IOException("Stream represented illegal ResultGroup!", wrongo);
        }
    }

    private void verify(){
        if(values == null){
            throw new IllegalStateException("Value records: null") ;
        }

        if(targets == null){
            throw new IllegalStateException("Target records: null");
        }
    }
}
